package com.flink.operator;

import com.flink.entity.User;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.LocalDate;

/**
 * 描述:
 * 算子运算操作Api
 *
 * @author yanzhengwu
 * @create 2022-07-21 9:28
 */
public class OperatorData {
    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);

        DataStreamSource<User> elementsStream = env.fromElements(
                new User("用户2", "家电", LocalDate.of(2022, 6, 2), 12, 10000L),
                new User("用户1", "零食", LocalDate.of(2022, 1, 1), 11, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户3", "椅子", LocalDate.of(2022, 6, 3), 14, 10000L),
                new User("用户3", "手机", LocalDate.of(2022, 6, 3), 15, 10000L),
                new User("用户1", "电脑", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户1", "手机", LocalDate.of(2022, 6, 16), 16, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 60000L)
        );

//        mapApi(elementsStream);
//        filterApi(elementsStream);
//        flatMapApi(elementsStream);
//        keyByApi(elementsStream);
//        reduceApi(elementsStream);
        richApi(elementsStream);
        //执行环境的执行入口
        env.execute();
    }

    /**
     * 收集所需要的data 可以参考java8的 stream.map
     *
     * @param data
     */
    static void mapApi(DataStreamSource<User> data) {
        //关于lambda函数接口的写法很多种 这里写的是最简单的函数引用;
        //数接口可以支持  ->函数, :: 函数,匿名类重写函数,实现类函数
        SingleOutputStreamOperator<String> map = data.map(User::getName);

        map.print("name->");


    }

    /**
     * 过滤所需要的data 可以参考java8的 stream.filter
     *
     * @param data
     */
    static void filterApi(DataStreamSource<User> data) {

        SingleOutputStreamOperator<User> user1 = data.filter(item -> item.getName().equals("用户1"));

        user1.print("用户1->");

    }

    /**
     * 扁平化所需要的data,可以理解为平铺数据； 可以参考java8的 stream.flatmap
     * 注意使用lambda表达式会有泛型擦除的坑爹操作，
     * 所以需要最后returns给予明确的返回值类型，否则报错，匿名类和实现类则没有这种问题
     *
     * @param data
     */
    static void flatMapApi(DataStreamSource<User> data) {

        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator =
                data.flatMap((User user, Collector<String> collector) -> {
                    //此方法块中也可以进行filter操作，处理比较灵活
                    //果判断逻辑代码比较多建议拆开使用，增加可读性
                    collector.collect(user.getName());

//                }).returns(TypeInformation.of(String.class));
                }).returns(Types.STRING);

        stringSingleOutputStreamOperator.print("user1->");

    }

    /**
     * 根据key分组聚合计算，
     * 注意算子maxby和max的区别
     * max只针对分组数据字段做更新返回的数据只有分组字段是最新数据，其他字段则为上一次计算数据
     * maxBy则是返回当前最近一次数据计算的所有最新值得全量；
     * TODO 所有的聚合操作  都必须先做keyBy数据分区
     * 运行数据看数据对比
     *
     * @param data
     */
    static void keyByApi(DataStreamSource<User> data) {
        SingleOutputStreamOperator<User> max =
                data.keyBy(User::getName).max("age");
        max.print("max->");

        SingleOutputStreamOperator<User> maxBy =
                data.keyBy(User::getName).maxBy("age");
        maxBy.print("maxBy->");


    }

    /**
     * 规约聚合
     *
     * @param data
     */
    static void reduceApi(DataStreamSource<User> data) {
        //1、先做reduce聚合各个用户点击商品的次数
        SingleOutputStreamOperator<Tuple2<String, Long>> reduce = data.map(item -> Tuple2.of(item.getName(), 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(tuple2 -> tuple2.f0)
//                .reduce((v1, v2) -> Tuple2.of(v1.f0, v1.f1 + v2.f1))
                .reduce(OperatorData::v1v2);
        reduce.print("maxUser->");
        //2、筛选最大点击量的数据
        //这里有一个比较重要的点 就是做reduce之前必须先做keyBy分组；
        //但是已经根据用户分组了，不能在基于二元组里面的f0和f1进行分组，
        //所以这里可以用一个其他的字符串进行所有数据分在同一组数据进行再次的reduce
        //TODO 这可以理解为一种骚操作，在一些必要的时候只能这么写，但是后果可能所有数据都压在了同一个slot里面执行造成资源浪费和数据倾斜
        //当前逻辑如果两个用户点击的次数值一样则以最新数据为准也就是近期点击达到数量的用户（reduce函数规则可以自己写）。
        reduce.keyBy(t -> "key")
                .reduce((v1, v2) -> v1.f1 > v2.f1 ? v1 : v2)
                .print("maxCount ->");
    }

    /**
     * 规约聚合如果逻辑比较复杂比如需要清洗筛选之类的操作
     * 建议拆分出另一个方法进行编写，提高代码可读性
     * TODO reduce操作相当于 sum min 等聚合操作的更抽象的一层，也可以理解为具体聚合操作的底层实现
     *
     * @param v1 上一次计算的二元组数据
     * @param v2 当前需要计算的二元组数据
     * @return
     */
    static Tuple2<String, Long> v1v2(Tuple2<String, Long> v1, Tuple2<String, Long> v2) {
        return Tuple2.of(v1.f0, v1.f1 + v2.f1);
    }

    /**
     * 富函数调用 可以满足单个函数的不足
     * 在富函数中可以拿到当前任务运行的一系列上下文对象和task信息
     * TODO 可以当做拦截器或者拦截任务日志进行使用
     * 当开启并行任务时则可以看到每个任务的上下文执行状态，也可以自己设置运行状态进行记录
     *
     * @param data
     */
    public static void richApi(DataStreamSource<User> data) {
        data.map(new RichMapper())
                .setParallelism(2)
                .print();


    }


    /**
     * 富函数类调用实现
     * 可以作为任务开启结束执行日志使用，记录任务执行状态是否有异常
     */
    public static class RichMapper extends RichMapFunction<User, String> {

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println("open声明周期被调用：" + getRuntimeContext().getIndexOfThisSubtask() + "号任务开始");
        }

        @Override
        public String map(User value) {
            return value.getName();
        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println("close声明周期被销毁：" + getRuntimeContext().getIndexOfThisSubtask() + "号任务结束");

        }


    }
}
